#pragma once
#include<iostream>
#include<signal.h>
#include<memory>
#include <functional>
#include <errno.h>
#include<mutex>
#include<cstring>
#include <sys/stat.h>
#include <sys/eventfd.h>
#include<vector>
#include <sys/types.h>
#include <sys/socket.h>
#include <unistd.h>
#include<typeinfo>
#include<string>
#include<assert.h>
 #include <fcntl.h>
#include<algorithm>
#include<thread>
#include <sys/epoll.h>
#include <arpa/inet.h>
#include<unordered_map>
#include<ctime>
#include <netinet/in.h>
#include<sys/timerfd.h>
#include<cstdint>
#include <atomic>     
#include <condition_variable>
using namespace std;
using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
const int defaultbuffersize = 1024;
#define MAX_LISTEN 1024 //backlog

#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL DBG
#define LOG(level,format,...) do{\
   if(level < LOG_LEVEL) break;\   
   time_t t = time(NULL);\
   struct tm* ltm = localtime(&t);\
   char temp[32] = {0};\
   strftime(temp,32,"%H:%M:%S",ltm);\
   fprintf(stdout,"[%p %s %s:%d] " format "\n",(void*)pthread_self(),temp,__FILE__,__LINE__,##__VA_ARGS__);\
}while(0)
#define INF_LOG(format,...) LOG(INF,format,##__VA_ARGS__)
#define DBG_LOG(format,...) LOG(DBG,format,##__VA_ARGS__)
#define ERR_LOG(format,...) LOG(ERR,format,##__VA_ARGS__)

class Buffer
{
private:
  vector<char> _buffer;//使用vector进行内存管理
  uint64_t _read_index;//读位置
  uint64_t _write_index;//写位置
 public:
   Buffer()
   :_buffer(defaultbuffersize),
   _read_index(0),
   _write_index(0)
   {}

   //提供一个公共的缓冲区起始位置: 获取起始地址还可以使用data() at(0) [0] front()
   char* Begin(){return &*_buffer.begin();}
   //当前读位置 
   char*  ReadPos(){return Begin() + _read_index;} 
   //当前写位置
   char*  WritePos(){return Begin() + _write_index;}
   //移动读指针
   void MoveReadIdx(uint64_t len)
   {
      if(len == 0) return;
      assert(len<=ReadAbleSize());//要读取长度不可超过可以读取数据大小
      _read_index += len;
   }
   //移动写指针
   void MoveWriteIdx(uint64_t len)
   {
      if(len == 0) return;
      //1.需要先保证后续空闲空间足够
      assert(TailSpace()>=len);
      //2.移动写指针
      _write_index += len;
   }
   //获取可读数据大小
   uint64_t ReadAbleSize(){return _write_index - _read_index;}
   //获取缓冲区末尾空闲空间大小---写偏移之后的空闲空间
   uint64_t TailSpace(){return  _buffer.size()-_write_index;}
   //获取缓冲区起始空闲空间大小 --- 读偏移之前的空闲空间
   uint64_t HeadSpace(){return _read_index;}
   //确保可写空间足够(移动+扩容)
   void EnsureWriteSpace(uint64_t len)
   {
      //1.先判断末尾空间大小是否足够
      if(len<=TailSpace()) return;
      //2.末尾空间不足，再加上起始空闲空间判断,此时向后移动
      if(len<=TailSpace()+HeadSpace())
      {
          uint64_t readsize = ReadAbleSize();//先保存可读数据位置
          std::copy(ReadPos(),ReadPos()+readsize,Begin());
          _read_index = 0;
          _write_index = readsize;
          return;
      }
      //3.起始+末尾空闲时间都不足,则扩容
      _buffer.resize(_write_index+len);
   }
   //写入数据
   void Write(const void*data,uint64_t len)
   {
      if(len == 0) return;
       //1.确保空间足够
       EnsureWriteSpace(len);
       //2.拷贝数据
       const char* d = (const char*)data;
       std::copy(d,d+len,WritePos());
   }
   //读取数据
   void Read(void* buf,uint64_t len)
   {
     //1.确保可读数据足够
     assert(len<=ReadAbleSize());
     //2.读取数据到buf
     std::copy(ReadPos(),ReadPos()+len,(char*)buf);
   }
   //清空缓冲区
   void clear()
   {
     _write_index = _read_index = 0;
   }
   //将string内容写入 
   void WriteString(const string& data)
   {
      Write(data.c_str(),data.size());
   }
   //将另一个Buffer对象内容写入
   void WriteBuffer(Buffer& data)
   {  
      Write(data.ReadPos(),data.ReadAbleSize());
   }
   //缓冲区数据当作string返回
   string ReadAsString(uint64_t len)
   {
       assert(len<=ReadAbleSize());//确保len不超过可读数据大小
       string str;
       str.resize(len);
       Read(&str[0],len);
       return str;
   }

   //读完并弹出
   void ReadAndPop(void*buf,uint64_t len)
   {  //读取
      Read(buf,len);
      //移动指针
      MoveReadIdx(len);
   }
   uint64_t Size()
   {
    return _buffer.size();
   }
   //读取到string并弹出
   string ReadStringAndPop(uint64_t len)
   {
     //读取到字符串
     string str = ReadAsString(len);
     MoveReadIdx(len);
     return str;
   }
   //写入数据并弹出
   void WriteAndPush(const void* data,uint64_t len)
   {
      Write(data,len);
      MoveWriteIdx(len);
   }
   //写入string并弹出
   void WriteStringPush(const string& data)
   {
      WriteString(data);
      MoveWriteIdx(data.size());
   }
   //写入另一个Buffer对象并弹出
   void WriteBufferPush(Buffer& data) 
   {
      WriteBuffer(data);
      MoveWriteIdx(data.ReadAbleSize());
   }
   //获取一行数据
   //1.获取换行符位置
   char* FindCRLF()
   {
     void* res = memchr(ReadPos(),'\n',ReadAbleSize());
     return (char*)res;
   }
   //2.获取一行
   string GetLine()
   {
     char* pos = FindCRLF();
     if(pos == nullptr) return "";
     //把换行也取出来
     return ReadAsString(pos-ReadPos()+1);
   }
   //3.获取一行并弹出
   string GetLineAndPop()
   {
      string str = GetLine();
      MoveReadIdx(str.size());
      return str;
   }
};

class Socket
{
private:
    int _sockfd;
public:
    Socket():_sockfd(-1){}
    ~Socket(){Close();}
    Socket(int fd):_sockfd(fd){}
    int Fd()
    {return _sockfd;}
    //创建套接字
    bool Create()
    {
       _sockfd = ::socket(PF_INET,SOCK_STREAM,0);
       if(_sockfd < 0)
       {
          ERR_LOG("创建套接字失败!\n");
          return false;
       }
       return true;
    }
    //绑定地址信息
    bool Bind(const string& ip,uint16_t port)
    {
      struct sockaddr_in addr;
      addr.sin_family = AF_INET;
      addr.sin_port = htons(port);
      addr.sin_addr.s_addr = inet_addr(ip.c_str());
      socklen_t len = sizeof(addr);
      int n = ::bind(_sockfd,(struct sockaddr*)&addr,len);
      if(n < 0)
      {
          ERR_LOG("绑定失败！\n");
          return false;
      }
      return true;
    }
    //开始监听
    bool listen(int backlog = MAX_LISTEN)
    {
       int n = ::listen(_sockfd,backlog);
       if(n < 0)
       {
         ERR_LOG("监听失败\n");
         return false;
       }
       return true;
    }
    //向服务器发起链接
    bool Connect(const string& ip,uint16_t port)
    {
       struct sockaddr_in addr = {0};
      addr.sin_family = AF_INET;
      addr.sin_port = htons(port);
      addr.sin_addr.s_addr = inet_addr(ip.c_str());
      socklen_t len = sizeof(addr);
      int n = ::connect(_sockfd,(struct sockaddr*)&addr,len);
      if(n < 0)
      {
          ERR_LOG("连接失败！");
          return false;
      }
      return true;
    }
    //获取新连接
    int Accept() //新连接直接返回fd方便后续操作
    {
       int newfd = ::accept(_sockfd,NULL,NULL);
       if(newfd < 0)
       {
         ERR_LOG("获取新连接失败\n");
         return -1;
       }
       return newfd ;
    }
    //关闭套接字
    void Close()
    {
      if(_sockfd != -1)
      {
        ::close(_sockfd);
        _sockfd = -1; 
      }
    }
    //创建一个服务端连接:默认非阻塞
    bool CreateServer(uint16_t port,const string& ip = "0.0.0.0",bool flag = 0) //服务器默认接收所有网卡的数据
    {
       //1.创建套接字 2.bind地址信息 3.监听 4.开启地址复用 5.设置非阻塞
       if(Create() == false) 
       {
          ERR_LOG("Create Error");
          return false;
       }
       if(flag) NonBlock();
       ReuseAddress();
       if(Bind(ip,port) == false)
       {
          ERR_LOG("Bind Error");
          return false;
       }
       if(listen() == false)
       {
         ERR_LOG("Listen Error");
         return false;
       }
       return true;
    }
    //创建一个客户端连接
    bool CreateClient(uint16_t port,const string& ip)
    {
       //1.创建套接字 2.连接
       if(Create() == false) return false;
       if(Connect(ip,port) == false) return false;
       return true;
    }
    //设置套接字选项 --- 开启地址端口重用
    void ReuseAddress()
    {
        int val = 1;
        setsockopt(_sockfd,SOL_SOCKET,SO_REUSEADDR,(void*)&val,sizeof(int));
        val = 1;
        setsockopt(_sockfd,SOL_SOCKET,SO_REUSEPORT,(void*)&val,sizeof(int));
    }
    //设置套接字阻塞属性
    void NonBlock()
    {
       int flag = ::fcntl(_sockfd,F_GETFL,0);
       ::fcntl(_sockfd,F_SETFL,flag|O_NONBLOCK);
    }
    //接收数据
    ssize_t Recv(void* buf,size_t len,int flag = 0) //缺省为0，默认阻塞读写
    {
        ssize_t ret = ::recv(_sockfd,buf,len,flag);
        if(ret <= 0)
        {   //EAGAIN:当前socket接收缓冲区中没有数据,在非阻塞情况下才有这个错误
            //EINTER:表示当前scoket的阻塞等待,被信号打断
           if(errno == EAGAIN || errno == EINTR)
              return 0;
           ERR_LOG("socket recv failed\n"); 
           return -1; 
        }
        return ret;
    }
    //发送数据 
    ssize_t Send(const void* buf,size_t len,int flag = 0)
    {
      if(len == 0) return 0;
      ssize_t ret = ::send(_sockfd,buf,len,flag);
      if(ret < 0)
      {
         if(errno == EINTR || errno == EAGAIN)
           return 0;
         ERR_LOG("scoket send error\n");
         return -1;
      }
      return ret;
    }
    //非阻塞接收数据
    ssize_t NonBlockRecv(void*buf,size_t len)
    {
       return Recv(buf,len,MSG_DONTWAIT);
    }
    //非阻塞发送数据
    ssize_t NonBlockSend(void*buf,size_t len)
    {
       return Send(buf,len,MSG_DONTWAIT);
    }
};

class Poller;
class EventLoop;
class Channel
{
private:
   int _fd;
   EventLoop* _loop;
   uint32_t _events;//当前链接需要监控的事件
   uint32_t _revnts; //当前链接触发的事件
   using EventCallBack = std::function<void()>;
   EventCallBack _write_callback; //可读事件回调
   EventCallBack _read_callback;  //可写事件回调
   EventCallBack _error_callback; //错误事件回调
   EventCallBack _close_callback; //关闭事件回调
   EventCallBack _any_callback;   //任意事件回调
public:
   Channel(EventLoop* loop,int fd = -1)
   :_fd(fd),
   _loop(loop),
   _events(0),
   _revnts(0)
   {}  
   int Fd()
   {return _fd;}
   uint32_t Events()//监控的事件
   {return _events;}
   bool ReadAble()//当前是否监控可读:判断对应事件的比特位是否为1
   {return _events&EPOLLIN; }
   bool WriteAble()//当前是否监控可写
   {return _events&EPOLLOUT;}
   void EnableRead()//启动读事件监控：将对应比特位置1
   {_events |= EPOLLIN;Update();}
   void EnableWrite() //启动写事件监控
   {_events |= EPOLLOUT;Update();}
   void DisableRead()//关闭读事件监控：将对应比特位置0
   {_events &= ~EPOLLIN;Update();}
   void DisableWrite() //关闭写事件监控
   {_events &= ~EPOLLOUT;Update();}
   void DisableAllEvent()// 关闭任意事件监控：全设置为0
   {_events = 0;Update();}
   void Update();
   void Remove();  //移除监控:目前无法实现,后面调用eventloop移除监控

   void SetReadCallBack(const EventCallBack& cb) //设置读事件回调
   {_read_callback = cb;}
   void SetWriteCallBack(const EventCallBack& cb) //设置读事件回调
   {_write_callback = cb;}
   void SetErrorCallBack(const EventCallBack& cb) //设置读事件回调
   {_error_callback = cb;}
   void SetCloseCallBack(const EventCallBack& cb) //设置读事件回调
   {_close_callback = cb;}
   void SetAnyCallBack(const EventCallBack& cb) //设置读事件回调
   {_any_callback = cb;}
   void HandlerEvent() //事件处理:一旦链接事件触发就调用该函数,自己触发了什么事件如何处理由自己决定
   {

      if(_revnts & EPOLLIN || _revnts & EPOLLRDHUP || _revnts & EPOLLPRI) //断开链接认为要求上层读取数据,关闭链接后没有数据发送了
      {
         if(_read_callback) _read_callback();
      }
      //有可能链接释放的事件,一次只处理一个因此使用else if
      if(_revnts & EPOLLOUT)
      {
         if(_write_callback) _write_callback();
      }
      else if(_revnts & EPOLLERR)
      {
         if(_error_callback) _error_callback();
      }
      else if(_revnts & EPOLLHUP)
      {
         if(_close_callback) _close_callback();
      }
      if(_any_callback) _any_callback();
   }

   void SetREvents(uint32_t revnts)
   {
      _revnts = revnts;
   }
};

#define MAX_EPOLLSIZE 1024
class Poller
{
private:
    int _epfd;
    struct epoll_event _events[MAX_EPOLLSIZE];
    unordered_map<int,Channel*> _channels; //一个fd对应一个channel
private: //为实现对外功能的私有接口
    bool HasChannel(Channel*channel) //一个Channel是否已经添加了事件监控
    {
       auto it = _channels.find(channel->Fd());
       if(it == _channels.end()) return false;
       return true;
    }
    void Update(Channel*channel,int op) //本质就是对epoll进行操作--epoll_ctl
    {
         struct epoll_event event;
         event.events = channel->Events();
         event.data.fd = channel->Fd();
         int ret = ::epoll_ctl(_epfd,op,channel->Fd(),&event);
         if(ret < 0)
         {
            ERR_LOG("epoll_ctrL error");
            return;
         }    
    }
public: //对外开放接口
    Poller()
    {
        _epfd = ::epoll_create(MAX_EPOLLSIZE);
        if(_epfd < 0)
        {
          ERR_LOG("Epoll Create Error");
          exit(1);
        }
    }
    //添加或修改监控事件
    void UpdateEvent(Channel* channel)
    {
       //1.先查询是否channel添加了事件监控
       bool ret = HasChannel(channel);
       if(ret == false) //不存在则添加
       { 
          _channels[channel->Fd()] = channel;
          Update(channel, EPOLL_CTL_ADD);
          return;
       }
       //存在则修改
       Update(channel,EPOLL_CTL_MOD);
    }   
    //移除监控事件
    void RemoveEvent(Channel* channel)
    {
      //1.从哈希表中移除
      auto it = _channels.find(channel->Fd());
      if(it != _channels.end())
          _channels.erase(it);
      //2.移出红黑树
      Update(channel,EPOLL_CTL_DEL);   
    }
    void Poll(vector<Channel*>* active) //开始监控返回活跃链接
    {
         int ret = ::epoll_wait(_epfd,_events,MAX_EPOLLSIZE,-1);//-1表示调用阻塞直到有事件发生
         if(ret < 0)
         {
            if(ret == EINTR) return;//信号打断
            ERR_LOG("epoll_wait error:%s\n",strerror(errno));
            abort();
         }   
         for(int i = 0; i < ret ; i++)
         {
              auto it = _channels.find(_events[i].data.fd);
              if(it == _channels.end())//说明channel管理出现问题
              {
                ERR_LOG("no channel");
                abort();
              }
              it->second->SetREvents(_events[i].events); //设置实际就绪事件
            //   if(_events[i].events & EPOLLIN)
            //   DBG_LOG("%d就绪了EPOLLIN事件",_events[i].data.fd);
              active->push_back(it->second); //放进活跃队列
         }
    }
};


//定时器任务对象
class TimeTask
{
private:
   uint64_t _id;//任务对象的id
   uint32_t _timeout;//定时任务的超时时间
   TaskFunc _task_cb; //定时器对象要执行的定制任务
   ReleaseFunc _release;//用于删除TimeWheel中保存的定时器对象信息
   bool _canceled;
 public:
  
    TimeTask(uint64_t id,uint32_t delay,const TaskFunc& cb)
    :_id(id),_timeout(delay),_task_cb(cb),_canceled(false)
    {}  
    //析构的时候执行定制任务并且销毁在定时轮中的该任务信息
    ~TimeTask()
    {
        if(!_canceled)
        _task_cb();
        _release();
    }
    //设置销毁定时器任务对象函数
    void SetRelease(const ReleaseFunc& cb)
    {
        _release = cb;
    }
    uint32_t DelayTime()
    {return _timeout;}

    void Cancel()
    {_canceled = true;}
};

class EventLoop;
//时间轮对象
class TimeWheel
{
private:
     using  WeakTask = std::weak_ptr<TimeTask>;
     using  PtrTask = std::shared_ptr<TimeTask>;
    
     int _tick;//表示执行定时任务的指针,走到哪里释放哪里,释放哪里相当于执行哪里的任务
     int _capacity;//相当于时间轮的一圈多少 --- 最大延迟时间
     vector<vector<PtrTask>> _wheel;//二维数组作为时间轮注意要比capacity后声明
     unordered_map<uint64_t,WeakTask> _timers; //注意这里需要使用WeakPtr是因为添加新的定时任务(shared_ptr<TimeTask>)再使用shared_Ptr会增加不必要的计数
     
     EventLoop* _loop; //事件循环
     int _timerfd;//定时器描述符
     unique_ptr<Channel> _timer_channel;//

     //从时间轮中删除定时任务
     void RemoveTimeTask(uint64_t id)
     {
         auto it = _timers.find(id);
         if(it != _timers.end())
            _timers.erase(it);
     }
private:
     static int CreateTimerFd()
     {
         //1.创建定时器
        int timerfd = ::timerfd_create(CLOCK_MONOTONIC,0);
        if(timerfd < 0)
        {
            ERR_LOG("timerfd create error!");
            abort();
        }
         //2.启动定时器
         struct itimerspec itime;
         itime.it_value.tv_sec = 1;//第一次超时
         itime.it_value.tv_nsec = 0;
         itime.it_interval.tv_sec = 1;//第一次超时之后的超时间隔
         itime.it_interval.tv_nsec = 0;
         int n = ::timerfd_settime(timerfd,0,&itime,NULL);
         if(n < 0)
         {
            ERR_LOG("start timer error!");
            abort();
         }
         return  timerfd;
     }
     
     int ReadTimerfd()
     {
        uint64_t times;
        int ret = ::read(_timerfd,&times,sizeof(times));//写入数据之前会阻塞没有数据
        if(ret < 0)
        {
           ERR_LOG("read error");;
           abort();
        }
        return times;
     }

     //定时器fd可读事件回调：1.读取定时器数据进行清0 2.执行一波所有的过期定时任务
     void TimerCb()
     {
         int times = ReadTimerfd();
         for(int i = 0 ; i < times ; i++)
             RunTimeTask();
     }

public:
    TimeWheel(EventLoop* loop)
    :_tick(0),_capacity(60),_wheel(_capacity),_loop(loop),
    _timerfd(CreateTimerFd()),_timer_channel(new Channel(loop,_timerfd))//注意成员声明顺序
    {
        _timer_channel->SetReadCallBack(std::bind(&TimeWheel::TimerCb,this));
        _timer_channel->EnableRead();
    }
    //在时间轮添加定时任务
    void TimerAdd(uint64_t id,uint32_t delay,const TaskFunc& cb);

    void TimerAddInloop(uint64_t id,uint32_t delay,const TaskFunc& cb)
    {
        //1.创建定时任务设置好删除函数
        PtrTask pt(new TimeTask(id,delay,cb));
        //注意:因为编译器不会将对象的成员函数隐式转换成函数指针，所以必须在TimeWheel::RemoveTimeTask前添加&；
        pt->SetRelease(std::bind(&TimeWheel::RemoveTimeTask,this,id));//注意这里需要使用bind这是因为RemoveTimeTask第一个参数是this,而且我们统一了ReleaseFunc
        cout << "Create Task.." << _wheel.size() << endl ;
        //2.存放到时间轮
        int pos = (_tick+delay)%_capacity;
        _wheel[pos].push_back(pt);
        _timers[id] = WeakTask(pt);
    }
    //对指定定时任务进行刷新/延迟

    void RefreshTask(uint64_t id);

    void RefreshTaskInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
         if(it == _timers.end())
            return; //没找到指定定时任务 返回
        PtrTask pt = it->second.lock();//相当于新建了一个shared_ptr共享计数
        int delay = pt->DelayTime();
        int pos = (_tick+delay)%_capacity;
        _wheel[pos].push_back(pt);
    }
    //指针移动执行任务
    void RunTimeTask()
    {
        _tick = (_tick+1)%_capacity;
        //走到哪释放哪
        _wheel[_tick].clear(); //清空指定位置的数组,就会把数组中保存的所有管理定时器对象的shared_ptr軽放掉
    }
    
    //取消定时任务

    void CancelTask(uint64_t id);

    void CancelTaskInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
         if(it == _timers.end())
            return; //没找到指定定时任务 返回
        PtrTask pt = it->second.lock();//相当于新建了一个shared_ptr共享计数
        if(pt) pt->Cancel();
    }

    //Class TimerWheel: 是否存在某个定时任务
    bool HasTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if(it == _timers.end()) return false;
        return true;
    }
};



class EventLoop
{
 private:
    std::thread::id _thread_id; //线程id
    int _eventfd; //eventfd唤醒事件监控可能导致的阻塞
    std::unique_ptr<Channel> _event_channel;//为eventfd创建一个channel进行事件监控
    Poller _poller;//进行所有描述符的事件监控

    using Functor = function<void()>;
    vector<Functor> _tasks; //任务池
    std::mutex _mtx; //保证任务队列线程安全的锁
    TimeWheel _timer_wheel;//定时器模块
    
    static int CreateEfd()//创建eventfd
    {
        int efd = ::eventfd(0,EFD_CLOEXEC|EFD_NONBLOCK); //initval flag
        if(efd < 0)
        {
            ERR_LOG("create eventfd error");
            abort();
        }
        return efd;
    }

    void ReadEvent()
    {
        uint64_t val = 0;
        ssize_t ret = read(_eventfd,&val,sizeof(val));
        if(ret < 0)
        {    //信号打断  无数据可读
           if(errno == EINTR || errno == EAGAIN) return;
           ERR_LOG("Read Eventfd Failed");
           abort();
        }
    }
    //其实就是向eventfd写入一个数据,达到事件通知的效果,触发可读事件
    void WeakEvent()
    {
         uint64_t val = 1;
        ssize_t ret = write(_eventfd,&val,sizeof(val));
        if(ret < 0)
        {    //信号打断  无数据可读
           if(errno == EINTR || errno == EAGAIN) return;
           ERR_LOG("Write Eventfd Failed");
           abort();
        }
    }

public:
   int EventFd(){return _eventfd;}
   EventLoop():_thread_id(std::this_thread::get_id()),_eventfd(CreateEfd()),
   _event_channel(new Channel(this,_eventfd)), //TODO Channel修改为eventloop
   _timer_wheel(this)
   {
       //给eventfd设置可读事件回调，读取事件通知次数(注意绑定的时候注意指定类域还有注意this)
       _event_channel->SetReadCallBack(std::bind(&EventLoop::ReadEvent,this));
       //开启可读监控
       _event_channel->EnableRead();
   }

   //交换任务时需要加锁保证任务队列的线程安全
   void RunAllTask() //执行任务池中所有任务
   {
       vector<Functor> func;
       {
          std::unique_lock<std::mutex> _lock(_mtx);
          func.swap(_tasks);
       }
       //执行任务
       for(const auto& f : func)
       {
           f();
       }
   }

   void RunInLoop(const Functor& cb) //判断将要执行的任务是否处于当前线程中，如果是则执行，不是则压入队列
   {
       if(isInLoop()) return cb();//将要执行的任务处于当前线程，直接返回
       return QueueInLoop(cb); //不是处于同一线程，压入线程池中
   }
   void QueueInLoop(const Functor& cb)//将操作压入任务池：需要加锁保证线程安全
   {
       {
         std::unique_lock<std::mutex> _lock(_mtx);
         _tasks.push_back(cb);
       }
      //唤醒可能因为事件没就绪,而导致的epoll阻塞
      //其实就是给eventfd写入一个数据,eventfd就会触发可读事件,此时epoll不会阻塞,执行RunAlltask()
      WeakEvent();
   }

   bool isInLoop()//用于判断当前线程是否是EventLoop对应的线程
   {
      return _thread_id == std::this_thread::get_id();
   }
   bool UpdateEvent(Channel* channel){_poller.UpdateEvent(channel);}//添加/修改描述符的事件监控
   void RemoveEvent(Channel* channel){_poller.RemoveEvent(channel);} //移除描述符的监控
   void Start() //三步走：1.事件监控 2.就绪事件处理 3.执行任务
   {  
      while(1)
      {
        //1.监控
        vector<Channel*> actives;
        _poller.Poll(&actives);
        //2.就绪事件处理
        for(const auto& a : actives)
        {
             a->HandlerEvent();
        }
        //3.执行任务
        RunAllTask();
      }
   }
    
   void TimerAdd(uint64_t id,uint32_t delay,const TaskFunc& cb);
   void RefreshTask(uint64_t id);
   void CancelTask(uint64_t id);
   bool HasTimer(uint64_t id);

   void AssertInLoop()
   {assert(_thread_id == std::this_thread::get_id());}
};

void Channel::Remove()
{
   _loop->RemoveEvent(this);
}

void Channel::Update()
{
   _loop->UpdateEvent(this);
}

void TimeWheel::TimerAdd(uint64_t id,uint32_t delay,const TaskFunc& cb)
{
     _loop->RunInLoop(std::bind(&TimeWheel::TimerAddInloop,this,id,delay,cb));
    return;
}

void TimeWheel::RefreshTask(uint64_t id)
{
   _loop->RunInLoop(std::bind(&TimeWheel::RefreshTaskInLoop,this,id));
   return;
}

void TimeWheel::CancelTask(uint64_t id)
{
   _loop->RunInLoop(std::bind(&TimeWheel::CancelTaskInLoop,this,id));
   return;
}


void EventLoop::TimerAdd(uint64_t id,uint32_t delay,const TaskFunc& cb)
{return _timer_wheel.TimerAdd(id,delay,cb);}
void EventLoop::RefreshTask(uint64_t id)
{return _timer_wheel.RefreshTask(id);}
void EventLoop::CancelTask(uint64_t id)
{return _timer_wheel.CancelTask(id);}
bool EventLoop::HasTimer(uint64_t id)
{return _timer_wheel.HasTimer(id);}


class Any
{
 private:   
   class holder
   {
     public:
       virtual ~holder(){}
       virtual const std::type_info& type()=0; //获取子类对象保存的数据类型
       //针对当前对象自身克隆出一个新子类对象,方便我们后续针对一个Any对象构造新的Any对象
       virtual holder* clone() = 0; 
   };

   template<class T>
   class placeholder : public holder
   {
     public:
       ~placeholder()
       {}
       placeholder(const T& val):_val(val){}
       virtual const std::type_info& type(){return typeid(_val);}
       virtual holder* clone() {return new placeholder<T>(_val);}
     public:
        T _val;  
   };

   holder* _content;
public:

   //交换子类对象
    Any& swap(Any& otehr)
    {
        std::swap(_content,otehr._content);
        return *this;
    }                                                                                                                                                     

    Any():_content(nullptr){};

    template<class T>
    Any(const T&val)
    :_content(new placeholder<T>(val))
    {}

    Any(const Any& other)
    :_content(other._content ? other._content->clone() : nullptr) //判断是否为空 不为空则clone
    {}

    ~Any()
    {delete _content;}
    
    template<class T>
    T* get() //返回子类对象保存的数据的指针
    {
        return &((placeholder<T>*)_content)->_val;
    }

    //为val构造一个临时的通用容器,然后与当前容器自身进行指针交换,临时对象释放的时候,原先保存的数据也就被释放
    template<class T>
    Any& operator=(const T& val)//返回值Any&是为了能连续赋值
    {
        Any(val).swap(*this);
        return *this;
    }
    Any& operator=(const Any& other)
    {
        // 通过外部other对象构造一个临时Any对象
         Any(other).swap(*this);
        return *this;
    }
    
};

//链接管理模块Connection
typedef enum {DISCONNECTED,CONNECTING,CONNECTED,DISCONNECTING} ConnStatu;
//DISCONNECTED -- 链接关闭状态
//CONNECTING -- 链接建立成功-待处理状态
//CONNECTED -- 链接建立完成，各种设置已完成，可以通信的状态
//DISCONNECTING --- 链接待关闭状态
class Conncetion;
using PtrConnection = std::shared_ptr<Conncetion>;//解决链接释放野指针(?)
class Conncetion : public enable_shared_from_this<Conncetion>
{
private:
   uint64_t _conn_id;//标识一个链接的id，也作为定时器事件id
   //套接字管理
   int _sockfd;//链接关联的文件描述符
   bool _enable_inactive_release;//链接释放启动非活跃链接销毁的判断标志,默认为false
   ConnStatu _statu;
   Socket _socket; //套接字操作管理
   //链接事件管理
   EventLoop* _loop;
   Channel _channel; //链接事件管理
   //缓冲区管理
   Buffer _in_buffer;//输入缓冲区:存放从socket中读取到的数据
   Buffer _out_buffer;//输出缓冲区:请求的接收处理上下文
   //协议上下文管理
   Any _context;//请求的接收处理上下文

   //回调函数管理(不同阶段回调)
   using ConnectedCallBack = std::function<void(const PtrConnection&)>;//连接建立成功后如何处理
   using MessageCallBack   = std::function<void(const PtrConnection&,Buffer*)>;//连接收到数据如何处理
   using CloseCallBack = std::function<void(const PtrConnection&)>;//连接关闭前如何处理
   using AnyCallBack = std::function<void(const PtrConnection&)>;//任意事件有没有处理由用户决定
   ConnectedCallBack _connected_cb;
   MessageCallBack _message_cb;
   CloseCallBack _close_cb;
   AnyCallBack _any_cb;
   CloseCallBack _server_closed_cb;//从服务器组件内移除链接管理信息
private:
//保证在EventLoop线程中执行
   void SendInLoop( Buffer& buf)
   {
      if(_statu == DISCONNECTED) return;
      //1.将数据放到缓冲区
      _out_buffer.WriteBufferPush(buf);
      //2.启动可写事件监控
      if(_channel.WriteAble() == false) _channel.EnableWrite();
   }



   void ShutDownInLoop()
   {
       _statu = DISCONNECTING; 
       //细节:在这里设置完DISCONNECTING之后,如果后续发送缓冲区还有数据开启写监控调用回调HandlerWrite发现
       //是DISCONNECTING就会调用ReleaseInLoop()
       //1.设置待关闭状态
       //2.判断接收缓冲区是否还有数据要处理
       if(_in_buffer.ReadAbleSize() > 0)
       {
          if(_message_cb) _message_cb(shared_from_this(),&_in_buffer);
       }
       //3.判断发送缓冲区是否还有数据
       if(_out_buffer.ReadAbleSize() > 0)
       {
          if(_channel.WriteAble() == false)//开启写监控就会调用HandlerWrite关闭链接
            _channel.EnableWrite();
            // return;
       }
       //没有直接关闭
      if(_out_buffer.ReadAbleSize() == 0)
        Release();
   }
   void EnableInactiveReleaseInLoop(int sec)
   {
      //1.设置标记位
      _enable_inactive_release = true;
      //2.延迟/添加定时任务
      if(_loop->HasTimer(_conn_id))
        return _loop->RefreshTask(_conn_id);
      _loop->TimerAdd(_conn_id,sec,std::bind(&Conncetion::Release,this));
   }

   void CancelInactiveReleaseInLoop()
   {
      //1.切换标记位
      _enable_inactive_release = false;
      //2.取消定时任务
      if(_loop->HasTimer(_conn_id))
         _loop->CancelTask(_conn_id);
   }

   void UpgradeInLoop(const Any& context,const ConnectedCallBack& conn,const MessageCallBack&
               msg,const CloseCallBack& closed,const AnyCallBack& event)
   {
         _context = context;
         _connected_cb = conn;
         _message_cb = msg;
         _close_cb = closed;
         _any_cb = event;
   }
   void ReleaseInLoop()//实际释放接口
   {
       //1.修改状态
       _statu = DISCONNECTED;
       //2.移除事件监控
       _channel.Remove();
       //3.关闭文件描述符
       _socket.Close();
       //4.看是否还存在定时任务需要移除
       if(_loop->HasTimer(_conn_id))
           CancelInactiveReleaseInLoop();
      //5.调用回调:注意为避免因移除服务器管理而导致连接释放而出错，因此先调用用户的 
      if(_close_cb) _close_cb(shared_from_this());
      if(_server_closed_cb) _server_closed_cb(shared_from_this());
   }

   void EstablishedInLoop()//连接获取之后进行设置
   {
      assert(_statu == CONNECTING);
      //1.修改连接状态
      _statu = CONNECTED;
      //2.开启读事件监控：开启之后可能立即会有事件触发,如果此时启动了非活跃链接销毁,就会刷新活跃延迟销毁任务执行
      //因此启动读事件监控应该在设置非活跃链接是否销毁之后进行
      _channel.EnableRead();
      //3.调用用户设置的回调
      if(_connected_cb)
      {
         _connected_cb(shared_from_this());
         // DBG_LOG("after coonnected_cb");
      }
   }


public:
   Conncetion(EventLoop* loop,uint64_t conn_id,int sockfd)
   :_conn_id(conn_id),_sockfd(sockfd),_enable_inactive_release(false),_statu(CONNECTING),
   _socket(sockfd),_loop(loop),_channel(loop,sockfd)
   {
        _channel.SetReadCallBack(std::bind(&Conncetion::HandlerRead,this));
        _channel.SetWriteCallBack(std::bind(&Conncetion::HandlerWrite,this));
        _channel.SetCloseCallBack(std::bind(&Conncetion::HandlerClose,this));
        _channel.SetErrorCallBack(std::bind(&Conncetion::HandlerError,this));
        _channel.SetAnyCallBack(std::bind(&Conncetion::HandlerAnyEvent,this));
   }
   ~Conncetion()
   {
      DBG_LOG("RELEASE CONNECTION:%p",this);
   }
   void Send(const char* data,size_t len)//发送数据，将数据发送到缓冲区，启动写事件监控
   {
       //bug:可能外界传入的data是个临时空间,我们现在只是把发送操作传入任务池,有可能并没被立即执行
       //因此有可能执行的时候,data指向的空间有可能已经被释放
       Buffer buf;
       buf.WriteAndPush(data,len);
      _loop->RunInLoop(std::bind(&Conncetion::SendInLoop,this,std::move(buf))); //使用move
   }

   void ShutDown()//提供给组件使用者的关闭接口--并不是实际关闭还需要判断是否有数据待处理
   {_loop->RunInLoop(std::bind(&Conncetion::ShutDownInLoop,this));}

   void EnableInactiveRelease(int sec)//启动非活跃销毁，并定义多长时间无通信是非活跃，添加定时任务
   {_loop->RunInLoop(std::bind(&Conncetion::EnableInactiveReleaseInLoop,this,sec));}
   
   void CancelInactiveRelease()//取消非活跃销毁
   {_loop->RunInLoop(std::bind(&Conncetion::CancelInactiveReleaseInLoop,this));}

   //切换协议 -- 重置上下文以及阶段性处理函数
   void Upgrade(const Any& context,const ConnectedCallBack& conn,const MessageCallBack&
               msg,const CloseCallBack& closed,const AnyCallBack& event)
   {
      //保证必须在eventLoop()线程中执行
      _loop->AssertInLoop();
      //应该立即执行不能压入队列：能走到这里说明就是同线程，就会直接执行任务
      _loop->RunInLoop(std::bind(&Conncetion::UpgradeInLoop,this,context,
                      conn,msg,closed,event));
   }

   void Release()
   {
       _loop->QueueInLoop(std::bind(&Conncetion::ReleaseInLoop,this));
   }
   
   void Established()
   {_loop->RunInLoop(std::bind(&Conncetion::EstablishedInLoop,this));}

public:   
   int Fd()//获取链接管理的文件描述符
   {return _sockfd;}
   int Id()//获取链接ID
   {return _conn_id;}
   bool Connected()//是否处于Connected状态
   {return _statu == CONNECTED;}
   void SetContext(const Any& context)//设置上下文
   {_context = context;}
   Any* GetContext()//获取上下文，返回指针
   {return &_context;}
   void SetConnectedCallBack(const ConnectedCallBack& cb)
   {_connected_cb = cb;}
   void SetMessageCallBack(const MessageCallBack& cb)
   {_message_cb = cb;}
   void SetCloseCallBack(const CloseCallBack& cb)
   {_close_cb = cb;}
   void SetAnyCallBack(const AnyCallBack& cb)
   {_any_cb = cb;}
   void SetSvrClosedCallBack(const CloseCallBack& cb)
   {_server_closed_cb = cb;}

public: //五个Channel事件回调
   void HandlerRead()//描述符可读事件触发的回调，接收socket数据放到接收缓冲区再调用message_cb
   {
      //1.非阻塞接收socket数据，放到缓冲区
      char buf[65536];
      int ret = _socket.NonBlockRecv(buf,65535);
      if(ret < 0)
      {
         //  DBG_LOG("ret < 0");
          //返回-1表示连接断开:出错不能直接关闭还要看缓冲区是否有数据要处理
          return ShutDownInLoop();
      }
      buf[ret] = 0;
      // DBG_LOG("%s",buf);
      //返回0表示未读取到数据,并不是连接断开
      //读取数据到缓冲区，并且移动写偏移
      _in_buffer.WriteAndPush(buf,ret);
      //2.调用用户message_cb
      if(_in_buffer.ReadAbleSize() > 0)
      {
         //  DBG_LOG("message_cb");
          _message_cb(shared_from_this(),&_in_buffer);
          return;
      }
   }

   void HandlerWrite()//描述符可写事件触发的回调，将发送缓冲区中数据进行发送
   {
      //1.将输出缓冲区中数据发送
      ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadPos(),_out_buffer.ReadAbleSize());
      if(ret < 0) //发送错误关闭连接
      {  //关闭连接之前查看接收缓冲区是否有数据要处理(出错也就不能发,查看发送缓冲区就没必要了)
         if(_in_buffer.ReadAbleSize() > 0)
            _message_cb(shared_from_this(),&_in_buffer);
         return Release();//直接关闭释放
      }
      //2.不要忘记移动读偏移(因为我们已经发送了)
      _out_buffer.MoveReadIdx(ret);
      if(_out_buffer.ReadAbleSize() == 0)
      {
          _channel.DisableWrite();//关闭写监控
          //如果当前连接时待关闭,有数据则处理数据(前面部分代码处理)，没有(或处理完了)则释放连接
          if(_statu == DISCONNECTING)
          return Release();
      }
   }

   void HandlerClose()//触发挂断事件
   {
      //输入缓冲区有数据就处理
      if(_in_buffer.ReadAbleSize() > 0)
        _message_cb(shared_from_this(),&_in_buffer);
      return Release();
   }

   void HandlerError()//触发出错事件
   {
      HandlerClose();
   }

   void HandlerAnyEvent()//触发任意事件
   {
      //1.刷新活跃度
      if(_enable_inactive_release)
          _loop->RefreshTask(_conn_id);
      //2.调用用户组件者设置的任意事件回调
      if(_any_cb) _any_cb(shared_from_this());
   }
};


class Acceptor
{
private:
   Socket _socket;//监听套接字
   EventLoop* _loop;//对监听套接字进行事件监控
   Channel _channel;//监听套接字事件管理

   using AcceptCallBack = std::function<void(int)>;
   AcceptCallBack _accept_cb;
private:
  void HandlerRead() //可读事件回调 --- 获取新连接,调用accept_cb进行新连接处理
  {
     //1.获取新连接
     int newfd = _socket.Accept();
     if(newfd < 0) return;
     //2.调用获取新连接回调
     if(_accept_cb) _accept_cb(newfd);
  }

  int CreateServer(int port)
  {
     bool ret = _socket.CreateServer(port);
     assert(ret == true);
     return _socket.Fd();
  }
  
public:
    
   int LstFd()
   {return _socket.Fd();}

   Acceptor(EventLoop* loop,int port)
   :_socket(CreateServer(port)),_loop(loop),_channel(loop,_socket.Fd())
   {
       _channel.SetReadCallBack(std::bind(&Acceptor::HandlerRead,this));
   }

   void Listen()
   {
       _channel.EnableRead();
   }

   void SetAcceptCallBack(const AcceptCallBack& cb)
   {_accept_cb = cb;}

};

//每个线程对应一个eventLoop
class LoopThread
{
private:
   //互斥锁&&条件变量:保证获取EventLoop对象指针的同步关系，防止获取时还没实例化
   std::mutex _mtx;
   std::condition_variable _cv;
   EventLoop* _loop; //EventLoop对象对应指针，需要在线程入口函数初始化
   std::thread _thread;//EventLoop对象对应线程
private:
   void ThreadEntry()
   {
      //不new对象的原因是想，EventLoop生命周期随线程
      EventLoop loop;
      {
         unique_lock<std::mutex> lck(_mtx);
         _loop = &loop;
         _cv.notify_all();//唤醒所有在条件变量上等的线程
      }
      //  while(1)
      //  {
      loop.Start();
      //  }
   } 

public:
   LoopThread()
   :_loop(nullptr),_thread(thread(&LoopThread::ThreadEntry,this))
   {}
   
   EventLoop* GetLoop()
   {
      EventLoop* loop = NULL;
      {
         unique_lock<std::mutex> lck(_mtx);
         _cv.wait(lck,[&](){
             return _loop != NULL;
         });//在条件变量上等EventLoop实例化
         loop = _loop;
      }
      return loop;
   }

   int Eventfd()
   {
       {
         unique_lock<std::mutex> lck(_mtx);
         _cv.wait(lck,[&](){
             return _loop != NULL;
         });//在条件变量上等EventLoop实例化
      }
      return _loop->EventFd();
   }

};

class LoopThreadPool
{
private:
    int _thread_count;//从属线程数量
    int _next_loopix; //分配eventLoop对象
    EventLoop* _base_loop;//主Reactor线程
    vector<LoopThread*> _threads;//管理LoopThread线程
    vector<EventLoop*> _loops;//管理EventLoop对象

public:
   LoopThreadPool(EventLoop* base_loop)
   :_base_loop(base_loop),_thread_count(0),_next_loopix(0)
   {}

   void SetThreadCount(int count)//设置从属线程数量
   {_thread_count = count;}

   void Create()//线程池初始化
   {
      if(_thread_count > 0)
      {
         _threads.resize(_thread_count);
         _loops.resize(_thread_count);
         for(int i = 0 ; i < _thread_count ; i++)
         {
            _threads[i] = new LoopThread();
            _loops[i] = _threads[i]->GetLoop();
            //细节:在线程入口函数内部实例化EventLoop对象之前GetLoop会阻塞
         }
      }
   }

   EventLoop* NextLoop()//给链接分配EventLoop对象进行事件监控
   {
       if(_thread_count == 0)
         return _base_loop;
      _next_loopix = (_next_loopix+1) % _thread_count;
      return _loops[_next_loopix];
   }
};

//TcpServer模块将前面的模块整合
class TcpServer
{
private:
   int _next_id; //一个自增长的链接id/定时任务id
   int _port;
   EventLoop _baseloop; //主Reactor线程
   Acceptor _acceptor; //监听套接字
   LoopThreadPool _pool;//eventLoop线程池
   unordered_map<uint64_t,PtrConnection> _conns;//保存管理的所有链接shared_Ptr对象
   int _timeout;//非活跃链接的超时时间
   bool _enable_inactive_release;//表示是否开启非活跃链接销毁
   
   //各种回调
   using Functor = function<void()>;
   using ConnectedCallBack = std::function<void(const PtrConnection&)>;//连接建立成功后如何处理
   using MessageCallBack   = std::function<void(const PtrConnection&,Buffer*)>;//连接收到数据如何处理
   using CloseCallBack = std::function<void(const PtrConnection&)>;//连接关闭前如何处理
   using AnyCallBack = std::function<void(const PtrConnection&)>;//任意事件有没有处理由用户决定
   ConnectedCallBack _connected_cb;
   MessageCallBack _message_cb;
   CloseCallBack _close_cb;
   AnyCallBack _any_cb;

 public:
    TcpServer(int port)
    :_next_id(0),_port(port),_acceptor(&_baseloop,port),_pool(&_baseloop),
    _enable_inactive_release(false)
    {
        _acceptor.SetAcceptCallBack(std::bind(&TcpServer::NewConnection,this,std::placeholders::_1));
        _acceptor.Listen();
    }
    void Start()//服务器启动
    {
       //1.初始化线程池
       _pool.Create();
       DBG_LOG("Create pool ok");
       //2.开启主线程的事件监控
       _baseloop.Start();
    }
    //回调函数设置   
   void SetConnectedCallBack(const ConnectedCallBack& cb)
   {_connected_cb = cb;}
   void SetMessageCallBack(const MessageCallBack& cb)
   {_message_cb = cb;}
   void SetCloseCallBack(const CloseCallBack& cb)
   {_close_cb = cb;}
   void SetAnyCallBack(const AnyCallBack& cb)
   {_any_cb = cb;}
   //添加定时任务
   void RunAfter(const Functor& task,int delay)
   {
      _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this,task,delay));
   }
   //启动非活跃超时链接销毁
   void EnableInactiveRelease(int timeout)
   {
      _timeout = timeout;
      _enable_inactive_release = true;
   }
   //用户设置线程池中线程数量
   void SetThreadCount(int count){_pool.SetThreadCount(count);}

private:
   //监听套接字Acceptror获取新链接之后的回调(给获取的链接创建Connection对象,设置事件回调,分配eventLoop对象监控,开启事件监控等)
   void NewConnection(int fd)
   {
     _next_id++;//自增链接id
     //分配线程池的线程给新链接,即分配eventLoop进行事件监控
     PtrConnection conn(new Conncetion(_pool.NextLoop(),_next_id,fd));
     cout << "gain a new link" << endl;
     conn->SetMessageCallBack(_message_cb);
     conn->SetConnectedCallBack(_connected_cb);
     conn->SetCloseCallBack(_close_cb);
     conn->SetAnyCallBack(_any_cb);
     conn->SetSvrClosedCallBack(std::bind(&TcpServer::RemoveConnection,this,std::placeholders::_1));
     if(_enable_inactive_release) conn->EnableInactiveRelease(_timeout); //启动-非活跃链接销毁
     //先启动非活跃连接销毁再启动读监控
     conn->Established();//就绪初始化:设置回调 启动读监控
   //   DBG_LOG("Established Success");
    _conns.insert(std::make_pair(_next_id,conn));//放进服务器组件管理
   }

   //移除链接的回调(给服务器组件设置的移除链接的回调,不从这删除,链接永远不会释放)
   void RemoveConnection(const PtrConnection& conn)
   {
      _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop,this,conn));
   }
   //保证线程安全
   void RunAfterInLoop(const Functor& task,int delay)
   {
      _next_id++;
      _baseloop.TimerAdd(_next_id,delay,task);
   }
   void RemoveConnectionInLoop(const PtrConnection& conn)
   {
      int id = conn->Id();
      auto it = _conns.find(id);
      if(it != _conns.end())
       _conns.erase(it);
   }

};


class NetWork
{
  public:
    NetWork()
    {
       DBG_LOG("SIGPIPE INIT");
       signal(SIGPIPE,SIG_IGN);
    }
};

static NetWork nw;